package com.jy.datapipeline.importserver.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.jy.datapipeline.common.util.AssertUtil;
import com.jy.datapipeline.common.constants.ExpConstants;
import com.jy.datapipeline.common.constants.PrefixConstants;
import com.jy.datapipeline.importserver.common.enums.impl.ReturnEnum;
import com.jy.datapipeline.common.exception.GlobalException;
import com.jy.datapipeline.common.util.PageUtil;
import com.jy.datapipeline.common.vo.ReturnVo;
import com.jy.datapipeline.importserver.entity.ImportSubtask;
import com.jy.datapipeline.importserver.mapper.ImportSubtaskMapper;
import com.jy.datapipeline.importserver.service.ImportRetrySubtaskService;
import com.jy.datapipeline.importserver.service.ImportSubTaskService;
import com.jy.datapipeline.importserver.service.ImportTaskLogService;
import com.jy.datapipeline.importserver.support.QuartzSupport;
import org.quartz.SchedulerException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @类名 ImportSubTaskServiceImpl
 * @描述 TODO
 * @作者 白条君
 * @创建日期 2023/4/23 15:09
 * @版本 1.0
 */
@Service
public class ImportSubTaskServiceImpl implements ImportSubTaskService {

    @Resource
    private ImportSubtaskMapper importSubtaskMapper;

    private final ImportTaskLogService importTaskLogService;

    private final ImportRetrySubtaskService importRetrySubtaskService;

    private final QuartzSupport quartzSupport;

    public ImportSubTaskServiceImpl(ImportTaskLogService importTaskLogService,
                                    ImportRetrySubtaskService importRetrySubtaskService,
                                    QuartzSupport quartzSupport) {
        this.importTaskLogService = importTaskLogService;
        this.importRetrySubtaskService = importRetrySubtaskService;
        this.quartzSupport = quartzSupport;
    }

    @Override
    public void addImportSubTask(ImportSubtask importSubTask) throws GlobalException {
        importSubTask.setCreateTime(new Date());
        int res = importSubtaskMapper.insertSelective(importSubTask);
        if (res == 0) {
            throw new GlobalException(ReturnEnum.SERVER_ERROR.getCode(), ExpConstants.ADD_FAIL);
        }
    }

    @Override
    public Set<Long> queryIds(Set<String> taskIds) {
        return importSubtaskMapper.selectIds(taskIds);
    }

    @Transactional
    @Override
    public void deleteSubTask(Long... ids) throws GlobalException, SchedulerException {
        Set<String> subtaskIds = importSubtaskMapper.selectSubtaskIdByIds(ids);

        //判断子任务是否存在log记录
        for (String subtaskId : subtaskIds) {
            Integer isExists = importTaskLogService.isExists(subtaskId);
            if (isExists != null) {
                importTaskLogService.deleteTaskLogsBySubtaskId(subtaskId);
            }
        }

        //删除重试任务
        Set<Long> idSet = new HashSet<>(Arrays.asList(ids));
        for (Long subtaskId : idSet) {
            quartzSupport.delete(PrefixConstants.SUBTASK_PREFIX + "_" + subtaskId, "import_subtask");
        }
        try {
            importRetrySubtaskService.deleteBySubtaskIds(idSet);
        } catch (Exception ignore) {}

        Integer resource = importSubtaskMapper.selectResource(ids);
        if (ObjectUtil.isNull(resource) || resource.equals(0)) {
            throw new GlobalException(ReturnEnum.SERVER_ERROR.getCode(), ExpConstants.RESOURCE_EXP);
        }

        int res = importSubtaskMapper.deleteByPrimaryKey(ids);
        if (res == 0) {
            throw new GlobalException(ReturnEnum.SERVER_ERROR.getCode(), ExpConstants.DELETE_FAIL);
        }
    }

    @Override
    public Integer getTryCount(long id) {
        return importSubtaskMapper.selctTryCount(id);
    }

    @Override
    public void modifySubTask(ImportSubtask importSubTask) throws GlobalException {
        int version = importSubtaskMapper.getVersion(importSubTask.getId());
        importSubTask.setVersion(version);

        int res = importSubtaskMapper.updateSubTask(importSubTask);
        if (res == 0) {
            throw new GlobalException(ReturnEnum.SERVER_ERROR.getCode(), ExpConstants.MODIFY_FAIL);
        }
    }

    @Override
    public void modifySubTasksStatus(List<ImportSubtask> importSubTaskList) {
        importSubtaskMapper.updateSubtaskStatus(importSubTaskList);
    }

    @Override
    public String queryTaskIdBySubtaskId(long id) {
        return importSubtaskMapper.selectTaskIdById(id);
    }

    @Override
    public Integer queryFailSubtask(String taskId) {
        return importSubtaskMapper.selectFailSubtask(taskId);
    }

    @Override
    public ReturnVo queryByPage(Long pageNo, Long pageSize, String subtaskStatus, String taskId, Integer trySubtaskLoadCurrentCount) {
        pageNo = (pageNo - 1) * pageSize;
        List<ImportSubtask> importSubtasks = importSubtaskMapper.selectByPage(pageNo, pageSize, subtaskStatus, taskId, trySubtaskLoadCurrentCount);
        if (ObjectUtil.isNotNull(importSubtasks) && importSubtasks.size() > 0) {
            importSubtasks = importSubtasks.stream()
                    .peek(importSubtask -> importSubtask.setTryCount(null))
                    .sorted((importSubtask1, importSubtask2) -> {
                        Date createTime1 = importSubtask2.getCreateTime();
                        Date createTime2 = importSubtask1.getCreateTime();
                        if (createTime1 != null && createTime2 != null) {
                            return createTime1.compareTo(createTime2);
                        }
                        return 0;
                    })
                    .collect(Collectors.toList());
            long totalSize = importSubtaskMapper.selectTotalSizeByTaskId(taskId, subtaskStatus, trySubtaskLoadCurrentCount);
            long totalPage = PageUtil.getTotalPage(totalSize, pageSize);
            return ReturnVo.ok().put(importSubtasks).put("totalSize", totalSize).put("totalPage", totalPage);
        }
        return ReturnVo.ok().put(importSubtasks);
    }

    @Override
    public List<ImportSubtask> querySubtasksById(Long id) {
        return importSubtaskMapper.selectSubtasksById(id);
    }

    @Override
    public ImportSubtask querySubtaskById(long id) {
        return importSubtaskMapper.selectSubtaskById(id);
    }

    @Override
    public void modifySubTaskRetryLoadCount(ImportSubtask importSubtask) throws GlobalException {
        int res = importSubtaskMapper.updateSubtaskRetryLoadCount(importSubtask);
        if (res == 0) {
            throw new GlobalException(ReturnEnum.SERVER_ERROR.getCode(), ExpConstants.MODIFY_FAIL);
        }
    }

    @Override
    public ImportSubtask querySubtasksBySubtaskId(Long id) {
        return importSubtaskMapper.selectSubtasks(id);
    }

    @Override
    public Integer getRetryLoadCurrentCount(Long id) {
        return importSubtaskMapper.selectRetryLoadCurrentCount(id);
    }

    @Override
    public List<ImportSubtask> queryNotStartSubtasks(String subtaskStatus) {
        return importSubtaskMapper.selectSubtasksByTaskStatus(subtaskStatus);
    }

    @Override
    public Integer getSubtaskStatus(long id) {
        return importSubtaskMapper.selectSubtaskStatus(id);
    }

    @Override
    public void batchRemove(Set<String> subtaskIds) {
        int size = importSubtaskMapper.batchDelete(subtaskIds);
        AssertUtil.totalAffectedRows(size, subtaskIds, ExpConstants.BATCH_DELETE_FAIL);
    }

    @Override
    public List<ImportSubtask> querySubtaskIds(Set<String> taskIds) {
        return importSubtaskMapper.selectSubtaskIds(taskIds);
    }

    @Override
    public ImportSubtask queryLatestStatus(String taskId) {
        return importSubtaskMapper.selectLatestStatus(taskId);
    }
}
